package com.atguigu.gmall.realtime.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.util.DruidDSUtil;
import com.atguigu.gmall.realtime.util.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * 将不同的维度数据写入到phoenix对应的表中
 *
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    //初始化连接池对象
    DruidDataSource dataSource = null;
    @Override
    public void open(Configuration parameters) throws Exception {
         dataSource = DruidDSUtil.createDataSource();
    }

    @Override
    public void invoke(JSONObject jsonObj, Context ctx) throws SQLException {
        //获取输出表名
        String sinkTable = jsonObj.getString("sink_table");
        //将表名数据删除
        jsonObj.remove("sink_table");

        Set<String> keySet = jsonObj.keySet();
        Collection<Object> value = jsonObj.values();
        //集合中的元素用逗号隔开
        StringUtils.join(keySet,",");
        StringUtils.join(value,",");

        //拼接sql语句
        String upsertSQL = "upsert into "+ GmallConfig.PHOENIX_SCHEMA + "." + sinkTable
                + "(" + StringUtils.join(keySet,",") + ") " +
                "values " + "('" +
                StringUtils.join(value,"','") + "')";

        System.out.println("插入到表的的语句为：" + upsertSQL);
        //从连接池获取连接对象
        //连接池获取的连接对象会自动提交
        Connection connection = dataSource.getConnection();

        PhoenixUtil.executeSql(connection,upsertSQL);

    }
}
